package io.jnet.server.write;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import com.jfireframework.baseutil.collection.buffer.DirectByteBuf;
import com.jfireframework.baseutil.concurrent.CpuCachePadingInt;
import io.jnet.common.bufstorage.BufStorage;
import io.jnet.server.api.ServerChannelContext;
import io.jnet.server.api.ServerListener;
import io.jnet.server.api.WriteHandler;

/**
 * 分离式写处理器。只关注获取数据并且写出。不会调用读取注册的功能
 * 
 * @author linbin
 *
 */
public class SeparationWriteHandler implements WriteHandler
{
    private final ByteBuf<?>[]              bufArray;
    private final static int                WORK             = 1;
    private final static int                IDLE             = 2;
    private final CpuCachePadingInt         status           = new CpuCachePadingInt(IDLE);
    private int                             currentSendCount = 0;
    private final int                       maxMerge;
    private final ServerChannelContext      serverChannelContext;
    private final AsynchronousSocketChannel socketChannel;
    private final ServerListener            channelListener;
    private final BufStorage                bufStorage;
    
    public SeparationWriteHandler(int maxMerge, AsynchronousSocketChannel socketChannel, ServerListener channelListener, BufStorage bufStorage, ServerChannelContext serverChannelContext)
    {
        bufArray = new ByteBuf<?>[maxMerge];
        this.maxMerge = maxMerge;
        this.serverChannelContext = serverChannelContext;
        this.socketChannel = socketChannel;
        this.channelListener = channelListener;
        this.bufStorage = bufStorage;
    }
    
    @Override
    public void completed(Integer result, ByteBuf<?> buf)
    {
        ByteBuffer buffer = buf.cachedNioBuffer();
        if (buffer.hasRemaining())
        {
            socketChannel.write(buffer, buf, this);
            return;
        }
        buf.release();
        channelListener.afterWrited(serverChannelContext, currentSendCount);
        writeNextBuf();
    }
    
    private void writeNextBuf()
    {
        currentSendCount = bufStorage.batchNext(bufArray, maxMerge);
        if (currentSendCount != 0)
        {
            int needSize = 0;
            for (int i = 0; i < currentSendCount; i++)
            {
                needSize += bufArray[i].remainRead();
            }
            ByteBuf<?> buf = DirectByteBuf.allocate(needSize);
            for (int i = 0; i < currentSendCount; i++)
            {
                buf.put(bufArray[i]);
                bufArray[i].release();
                bufArray[i] = null;
            }
            socketChannel.write(buf.cachedNioBuffer(), buf, this);
        }
        else
        {
            status.set(IDLE);
            if (bufStorage.isEmpty() == false)
            {
                commitWrite();
            }
        }
    }
    
    @Override
    public void failed(Throwable exc, ByteBuf<?> buf)
    {
        buf.release();
        channelListener.catchException(exc, serverChannelContext);
    }
    
    public void commitWrite()
    {
        int now = status.value();
        if (now == IDLE && status.compareAndSwap(IDLE, WORK))
        {
            writeNextBuf();
        }
    }
}
